package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo2SinkMysql {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.readTextFile("flink/data/students.txt")
    ds.print()

    ds.addSink(new MysqlSink)

    env.execute()

  }


}


class MysqlSink extends RichSinkFunction[String] {
  var conn: Connection = _

  /**
    * 创建网络连接
    */
  override def open(parameters: Configuration): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
    conn = DriverManager.getConnection("jdbc:mysql://master:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "123456")

  }

  override def close(): Unit = {
    conn.close()
  }

  /**
    * invoke 方法，每一条数据都会执行异常
    *
    */

  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {

    val split: Array[String] = value.split(",")

    val stat: PreparedStatement = conn.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

    stat.setString(1, split(0))
    stat.setString(2, split(1))
    stat.setInt(3, split(2).toInt)
    stat.setString(4, split(3))
    stat.setString(5, split(4))

    stat.executeUpdate()

  }
}

